package com.ztgf.search.server.service;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Maps;
import com.ztgf.search.models.annotation.ESMappingField;
import com.ztgf.search.models.annotation.ESMappingType;
import com.ztgf.search.models.service.ElasticsearchService;
import com.ztgf.search.server.utils.StringHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboService;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.admin.indices.alias.Alias;
import org.elasticsearch.action.admin.indices.alias.IndicesAliasesRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.ReindexRequest;
import org.elasticsearch.index.reindex.ScrollableHitSource;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;

import java.io.IOException;
import java.lang.reflect.Field;
import java.util.List;
import java.util.Map;

@Slf4j
@DubboService
public class ElasticsearchServiceImpl implements ElasticsearchService {

    @Qualifier("restHighLevelClient")
    @Autowired
    public RestHighLevelClient client;

    public static final String simplePatternSplitSymbol = "comma,underline";

    private static final String searchKeyAlias = "search_key";

    private static final String searchKeyType = "keysval";

    @Override
    public boolean createIndex(String indexName, String aliasename, Integer shards, Integer replicas) {
        CreateIndexRequest request = new CreateIndexRequest(indexName);
        if (StringHelper.isNotEmpty(aliasename)) {
            request.alias(new Alias(aliasename));
        }
        request.timeout(TimeValue.timeValueMinutes(1));
        request.masterNodeTimeout(TimeValue.timeValueMinutes(2));
        try {
            CreateIndexResponse createIndexResponse = client.indices().create(request, RequestOptions.DEFAULT);
            boolean acknowledged = createIndexResponse.isAcknowledged();
            boolean shardsAcknowledged = createIndexResponse.isShardsAcknowledged();
            if (acknowledged && shardsAcknowledged) {
                log.info("索引创建成功");
            }
            return acknowledged && shardsAcknowledged;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public boolean addMapping(String indexName, String typeName, Class<?> clazz) {
        if (clazz.getAnnotation(ESMappingType.class) == null) {
            return true;
        }
        PutMappingRequest putMappingRequest = new PutMappingRequest(indexName);
        putMappingRequest.type(typeName);

        Map<String, Object> jsonMap = Maps.newHashMap();
        Map<String, Object> properties = Maps.newHashMap();

        Field[] fields = clazz.getDeclaredFields();
        for (Field f : fields) {
            ESMappingField esMappingField = f.getAnnotation(ESMappingField.class);
            if (esMappingField != null) {
                String fieldname = esMappingField.fieldName();
                String datatype = esMappingField.dataType();
                String analyzer = esMappingField.setAnalyzer();
                boolean isanalye = esMappingField.isAnalyze();

                Map<String, Object> m = Maps.newHashMap();
                m.put("type", datatype);
                if (isanalye && StringHelper.isNotEmpty(analyzer)) {
                    m.put("analyzer", analyzer);
                    m.put("search_analyzer", analyzer);
                }
                properties.put(fieldname, m);
            }
        }
        jsonMap.put("properties", properties);
        putMappingRequest.source(jsonMap);
        putMappingRequest.timeout(TimeValue.timeValueMinutes(2));
        try {
            AcknowledgedResponse putMappingResponse = client.indices().putMapping(putMappingRequest,
                    RequestOptions.DEFAULT);
            return putMappingResponse.isAcknowledged();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public boolean deleteIndex(String indexName) {
        DeleteIndexRequest request = new DeleteIndexRequest(indexName);
        request.timeout(TimeValue.timeValueMinutes(2));
        request.masterNodeTimeout(TimeValue.timeValueMinutes(3));
        try {
            AcknowledgedResponse deleteIndexResponse = client.indices().delete(request, RequestOptions.DEFAULT);
            return deleteIndexResponse.isAcknowledged();
        } catch (Exception e) {
            log.info("删除索引错误信息:{}", e.getMessage());
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public void reindex(String old_indexname, String new_indexname) {
        ReindexRequest request = new ReindexRequest();
        request.setSourceIndices(old_indexname);
        request.setDestIndex(new_indexname);
        request.setDestVersionType(VersionType.EXTERNAL);
        request.setSourceBatchSize(1000);
        request.setDestOpType("create");
        request.setConflicts("proceed");
        request.setScroll(TimeValue.timeValueMinutes(10));
        request.setTimeout(TimeValue.timeValueMinutes(20));
        request.setRefresh(true);
        try {
            BulkByScrollResponse bulkResponse = client.reindex(request, RequestOptions.DEFAULT);
            boolean timedOut = bulkResponse.isTimedOut();
            long totalDocs = bulkResponse.getTotal();
            long updatedDocs = bulkResponse.getUpdated();
            long createdDocs = bulkResponse.getCreated();
            long deletedDocs = bulkResponse.getDeleted();
            long batches = bulkResponse.getBatches();
            long noops = bulkResponse.getNoops();
            long versionConflicts = bulkResponse.getVersionConflicts();
            long bulkRetries = bulkResponse.getBulkRetries();
            long searchRetries = bulkResponse.getSearchRetries();
            TimeValue throttledMillis = bulkResponse.getStatus().getThrottled();
            TimeValue throttledUntilMillis = bulkResponse.getStatus().getThrottledUntil();
            List<ScrollableHitSource.SearchFailure> searchFailures = bulkResponse.getSearchFailures();
            List<BulkItemResponse.Failure> bulkFailures = bulkResponse.getBulkFailures();

            log.info("timedOut: " + timedOut);
            log.info("totalDocs: " + totalDocs);
            log.info("updatedDocs: " + updatedDocs);
            log.info("createdDocs: " + createdDocs);
            log.info("deletedDocs: " + deletedDocs);
            log.info("batches: " + batches);
            log.info("noops: " + noops);
            log.info("versionConflicts: " + versionConflicts);
            log.info("bulkRetries: " + bulkRetries);
            log.info("searchRetries: " + searchRetries);
            log.info("throttledMillis: " + throttledMillis.toString());
            log.info("throttledUntilMillis: " + throttledUntilMillis.toString());
            log.info("searchFailures: " + searchFailures.size());
            log.info("bulkFailures: " + bulkFailures.size());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public boolean changeAliasAfterReindex(String aliasname, String old_indexname, String new_indexname) {
        IndicesAliasesRequest.AliasActions addIndexAction = new IndicesAliasesRequest.AliasActions(
                IndicesAliasesRequest.AliasActions.Type.ADD).index(new_indexname).alias(aliasname);
        IndicesAliasesRequest.AliasActions removeAction = new IndicesAliasesRequest.AliasActions(
                IndicesAliasesRequest.AliasActions.Type.REMOVE).index(old_indexname).alias(aliasname);

        IndicesAliasesRequest indicesAliasesRequest = new IndicesAliasesRequest();
        indicesAliasesRequest.addAliasAction(addIndexAction);
        indicesAliasesRequest.addAliasAction(removeAction);
        try {
            AcknowledgedResponse indicesAliasesResponse = client.indices().updateAliases(indicesAliasesRequest,
                    RequestOptions.DEFAULT);
            boolean acknowledged = indicesAliasesResponse.isAcknowledged();
            if (acknowledged) {
            }
            return acknowledged;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 判断新增的记录是否已存在
     *
     * @return
     * @throws IOException
     */
    @Override
    public boolean isExist(String aliasName, String typeName, String id) {
        GetRequest getRequest = new GetRequest(aliasName, typeName, id);
        getRequest.fetchSourceContext(new FetchSourceContext(false));
        getRequest.storedFields("_none_");
        try {
            return client.exists(getRequest, RequestOptions.DEFAULT);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public boolean deleteData(String aliasName, String typeName, String id) {
        DeleteRequest deleteRequest = new DeleteRequest(aliasName, typeName, id);
        deleteRequest.timeout(TimeValue.timeValueMinutes(20));
        deleteRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
        deleteRequest.setRefreshPolicy("wait_for");
        try {
            DeleteResponse deleteResponse = client.delete(deleteRequest, RequestOptions.DEFAULT);
            RestStatus status = deleteResponse.status();
            log.info("删除es某条数据状态值 : {}", status.getStatus());
        } catch (Exception e) {
            e.printStackTrace();
            log.info("更新es删除某条数据失败!");
        }
        return true;
    }

    @Override
    public <T> Boolean saveOrUpdateEntity(T entity, String aliasName, String typeName, String id) {
        try {
            if (StringHelper.isNotEmpty(id)) {
                if (this.isExist(aliasName, typeName, id)) {
                    UpdateRequest updateRequest = new UpdateRequest(aliasName, typeName, id);
                    updateRequest.doc(JSONObject.toJSONString(entity), XContentType.JSON);
                    UpdateResponse updateResponse = client.update(updateRequest, RequestOptions.DEFAULT);
                    if (updateResponse.getResult() == DocWriteResponse.Result.CREATED) {
                        log.info("CREATED！");
                    } else if (updateResponse.getResult() == DocWriteResponse.Result.UPDATED) {
                        log.info("UPDATED！");
                    } else if (updateResponse.getResult() == DocWriteResponse.Result.DELETED) {
                        log.info("DELETED！");
                    } else if (updateResponse.getResult() == DocWriteResponse.Result.NOOP) {
                        log.info("NOOP！");
                    }
                    return true;
                } else {
                    String jsonString = JSONObject.toJSONString(entity);
                    IndexRequest indexRequest = new IndexRequest(aliasName, typeName, id);
                    indexRequest.source(jsonString, XContentType.JSON);
                    IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
                    if (indexResponse != null && "CREATED".equals(indexResponse.getResult().toString())) {
                        log.info("CREATED！");
                        return true;
                    }
                }
            } else {
                String jsonString = JSONObject.toJSONString(entity);
                IndexRequest indexRequest = new IndexRequest(aliasName, typeName);
                indexRequest.source(jsonString, XContentType.JSON);
                IndexResponse indexResponse = client.index(indexRequest, RequestOptions.DEFAULT);
                if (indexResponse != null && "CREATED".equals(indexResponse.getResult().toString())) {
                    log.info("CREATED！");
                    return true;
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public Map<String, Long> statisticSearchKey(Integer returnDataSize) {
        Map<String, Long> map = Maps.newHashMap();
        SearchRequest searchRequest = new SearchRequest(searchKeyAlias);
        searchRequest.types(searchKeyType);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.aggregation(AggregationBuilders.terms("aggregationList").field("searh_key")
                .size(returnDataSize));
        searchRequest.source(searchSourceBuilder);
        try {
            SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            ParsedStringTerms terms = searchResponse.getAggregations().get("aggregationList");
            for (Terms.Bucket bucket : terms.getBuckets()) {
                map.put(bucket.getKeyAsString(), bucket.getDocCount());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return map;
    }

    @Override
    public <T> T getEntity(String aliasname, String type, String id, Class<T> clazz) {
        GetRequest getRequest = new GetRequest(aliasname, type, id);
        try {
            GetResponse getResponse = client.get(getRequest, RequestOptions.DEFAULT);
            if (getResponse.isExists()) {
                return JSONObject.parseObject(JSONObject.toJSONString(getResponse.getSourceAsMap()), clazz);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public boolean deleteBatchData(String aliasName, String typeName, String[] ids) {
        BulkRequest bulkRequest = new BulkRequest();
        for (String id : ids) {
            log.info("id-----------------------------------------"+id);
            DeleteRequest del = new DeleteRequest(aliasName, typeName, id);
            bulkRequest.add(del);
        }
        bulkRequest.timeout(TimeValue.timeValueMinutes(40));
        try {
            BulkResponse bulkResponse = client.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (bulkResponse.hasFailures()) {
            } else {
                return true;
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }
}
